// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/app/driver.h"

#include <cstdlib>
#include <exception>
#include <string>

#include <spdlog/async.h>
#include <spdlog/sinks/rotating_file_sink.h>

#include "backward.hpp"
#include "coin2/base/config/app_config.h"
#include "coin2/base/log.h"
#include "coin2/exchange/base/feed/ws/feed_subscription_spec.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/strategy/strategy.h"

namespace coin2::app {
namespace impl {

Driver* Driver::driver_for_signal_ = nullptr;
bool Driver::is_handling_signal_ = false;

void Driver::OnSignal(int signum) {
  if (!driver_for_signal_) {
    ::exit(signum);
    return;
  }

  if (is_handling_signal_) {
    return;
  }
  is_handling_signal_ = true;

  bool is_live = false;
  for (const auto& os : driver_for_signal_->order()->GetAllSystems()) {
    if (!os->is_sim_os()) {
      is_live = true;
      break;
    }
  }

  if (is_live) {
    driver_for_signal_->PostTask(std::bind(Driver::OnSignalImpl, signum, is_live));
  } else {
    OnSignalImpl(signum, is_live);
  }
}

void Driver::OnSignalImpl(int signum, bool is_live) {
  LOG(INFO) << "Catched signal [" << signum << "], gracefully exiting";
  driver_for_signal_->immediate_exit_on_signal_ = true;
  auto* strategy = driver_for_signal_->strategy_;
  if (strategy != nullptr) {
    strategy->onCleanup(driver_for_signal_);
  }
  if (!is_live || driver_for_signal_->immediate_exit_on_signal_) {
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    exit(signum);
  } else {
    driver_for_signal_->exitfunc_ = std::bind(::exit, signum);
  }
}

void Driver::Exit() {
  if (exitfunc_) {
    exitfunc_();
  } else {
    exit(0);
  }
}

void Driver::RedirectLogToFile(
    const std::string& log_file_path,
    int log_level,
    bool truncate,
    bool is_rotating) {
  // for now, only support spdlog
  if (log_file_path.empty()) {
    return;
  }
  if (log_level < 0 || log_level > 6) {
    return;
  }
  if (is_rotating) {
    // split log files every 1 GiB / at most 5 files
    auto rotating_logger = spdlog::rotating_logger_mt<spdlog::async_factory>(
        "file_logger",
        log_file_path,
        1048576 * 1024,
        4);
    spdlog::set_default_logger(rotating_logger);
    // flush every 5 sec
    spdlog::flush_every(std::chrono::seconds(5));
  } else {
    auto file_logger = spdlog::basic_logger_mt("file_logger", log_file_path, truncate);
    spdlog::set_default_logger(file_logger);
  }
  // SPDLOG_LEVEL_TRACE 0
  // SPDLOG_LEVEL_DEBUG 1
  // SPDLOG_LEVEL_INFO 2
  // SPDLOG_LEVEL_WARN 3
  // SPDLOG_LEVEL_ERROR 4
  // SPDLOG_LEVEL_CRITICAL 5
  // SPDLOG_LEVEL_OFF 6
  spdlog::flush_on(static_cast<spdlog::level::level_enum>(log_level));
  spdlog::info("log sink is redirecting to file {}", log_file_path);
}

Driver::Driver(
    DriverConfig config,
    coin2::strategy::IStrategy* strategy,
    exchange::di::ConstantInjectionBuilder ci_builder)
    : ci_(ci_builder),
      config_(ConsolidateDriverConfig(ci_, &config)),
      live_executor_(config.executor().live()),
      archive_executor_(config.executor().archive()),
      strategy_(strategy),
      product_holder_manager_(coin::proto::ProductHolderManagerConfig()),
      feed_(
          ci_,
          config.feed(),
          &live_executor_,
          &archive_executor_,
          static_cast<exchange::base::feed::IFeedSubscriber*>(strategy)),
      strategy_mgr_(
          ci_,
          config.strategy(),
          coin2::base::config::GetEnclosingInterval(config).first),
      order_(
          ci_,
          config.order(),
          &live_executor_,
          &archive_executor_,
          &product_holder_manager_,
          &feed_,
          static_cast<exchange::base::order::IOrderSubscriber*>(strategy),
          config.strategy().strategy_name()) {
  // controller_.reset(new Controller(live_executor_.MainIoc(), config.strategy().strategy_name()));
  // controller_->ExportMethods(order_.ExportMethods());
  boost::posix_time::nanoseconds interval(config.log().health_report_period_ns());
  health_report_timer_ =
      std::make_unique<boost::asio::deadline_timer>(*live_executor_.RawIoc(), interval);
  cleanup_timer_ =
      std::make_unique<boost::asio::deadline_timer>(*live_executor_.RawIoc(), interval);
}

void Driver::Run(bool verbose) {
  if (verbose) {
    LOG(INFO) << "[DriverConfig]\n" << config_.DebugString();
  }

  ::std::set_terminate([] {
    ::backward::StackTrace st;
    st.load_here(32);
    ::backward::Printer p;
    p.object = true;
    p.color_mode = ::backward::ColorMode::always;
    p.address = true;
    p.print(st, stderr);
    ::std::abort();
  });

  backward::SignalHandling sh;
  strategy_mgr_.Init(&order_);  // init oe.set_order_system()
  strategy_->Init(this);

  if ((live_executor_.NumSystems() != 0U) && (archive_executor_.NumSystems() != 0U)) {
    THROW() << "Both live and archive subsystem co-exist";
  } else if (live_executor_.NumSystems() != 0U) {
    // schedule health check timer
    if (config_.log().health_report_period_ns() > 0) {
      health_report_timer_->async_wait(std::bind(&Driver::LogHealthReport, this));
    }
    live_executor_.Start();
  } else if (archive_executor_.NumSystems() != 0U) {
    archive_executor_.Run();
  }
}

void Driver::RegisterCleanupTimeout(int64_t auto_shutdown_sec) {
  cleanup_timer_->expires_from_now(
      boost::posix_time::nanoseconds(auto_shutdown_sec * 1'000'000'000L));
  cleanup_timer_->async_wait(std::bind(Driver::OnSignal, SIGINT));
}

void Driver::GetHealthReport(int indent, std::string* report) const {
  // [HealthReport]
  //     [Feed]
  //         - Huobi (OK)
  //         - Okex (STARTING, uptime=23h)
  //               a1 (..)
  for (int i = 0; i < indent * 4; i++) {
    *report += ' ';
  }
  *report += "\n[Health Report]\n";  // preceeding newline for clean log
  feed_.GetHealthReport(indent + 1, report);
  order_.GetHealthReport(indent + 1, report);
  // strategy_->GetHealthReport(indent + 1, report);
}

void Driver::LogHealthReport() {
  std::string report;
  GetHealthReport(0, &report);
  LOG(INFO) << report;

  if (config_.log().health_report_period_ns() > 0) {
    health_report_timer_->expires_from_now(
        boost::posix_time::nanoseconds(config_.log().health_report_period_ns()));
    health_report_timer_->async_wait(std::bind(&Driver::LogHealthReport, this));
  }
}

void Driver::SetImmediateExitOnSignal(bool flag) { immediate_exit_on_signal_ = flag; }

void Driver::PostTask(std::function<void(void)> task) { live_executor_.MainIoc()->post(task); }

// TODO(xguo) Handle multiple delay tasked better. right now only one at a time.
void Driver::PostDelayedTask(std::function<void(void)> task, int64_t delay) {
  if (!scheduler_) {
    scheduler_.reset(new Scheduler(live_executor_.MainIoc()));
    scheduler_->SetCheckPeriod(10'000'000LL);
    scheduler_->Start();
  }
  scheduler_->CallLater(task, delay, "random");
}

void Driver::PostPeriodicTask(std::function<void(void)> task, int64_t period, int64_t clock) {
  if (!scheduler_) {
    scheduler_.reset(new Scheduler(live_executor_.MainIoc()));
    scheduler_->SetCheckPeriod(clock);
    scheduler_->Start();
  }
  scheduler_->AddPeriodicTask(task, period, "random");
}

}  // namespace impl

using coin2::app::DriverConfig;
using coin2::exchange::base::market::MarketExchangeApi;

// Make sure this method is idempotent, e.g. okay to call multiple times
DriverConfig& ConsolidateDriverConfig(
    const exchange::di::ConstantInjection& ci,
    DriverConfig* driver) {
  if (driver->is_conslidated_do_not_set_this_field()) {
    return *driver;
  }

  // merge common to exchanges
  if (driver->has_feed()) {
    auto& feed = *driver->mutable_feed();

    if (driver->has_alarm()) {
      for (int i = 0; i < driver->alarm().alarms_size(); ++i) {
        if (driver->alarm().alarms(i).add_feed1()) {
          std::string alarm11_name = fmt::format("alarm{}_{}", i, 1);
          feed.mutable_exchanges()->insert({alarm11_name, driver->alarm().alarms(i).product1()});
        }
        if (driver->alarm().alarms(i).add_feed2()) {
          std::string alarm12_name = fmt::format("alarm{}_{}", i, 2);
          feed.mutable_exchanges()->insert({alarm12_name, driver->alarm().alarms(i).product2()});
        }
      }
    }

    // 1. overwrite common to exchanges
    if (feed.has_common()) {
      for (auto& [name, config] : *feed.mutable_exchanges()) {
        auto c = feed.common();
        c.MergeFrom(config);
        config = c;
      }
    }

    // 2. fill system.mea with system.name
    for (auto& [name, config] : *feed.mutable_exchanges()) {
      if (!config.has_mea()) {
        config.set_mea(name);
      }
      CHECK_GT_THROW(name.size(), 0);
      CHECK_GT_THROW(config.mea().size(), 0);
    }

    // 3. expand feed recipes
    for (auto& [name, config] : *feed.mutable_exchanges()) {
      (void)name;

      // no channel => we use recipe
      if (config.channels_size() == 0) {
        auto mea = MarketExchangeApi::FromString(config.mea());
        const auto& consts = ci.GetFeedSubscriptionSpec(mea);

        // no recipe => default recipe
        if (!config.has_recipe()) {
          CHECK_GT_THROW(consts.default_recipe.size(), 0)
              << "You provided no 'recipe' or 'channels', and " << config.mea()
              << " has no default_recipe";
          LOG(INFO) << "The feed " << config.mea()
                    << " fallbacks to the default recipe: " << consts.default_recipe;
          config.set_recipe(consts.default_recipe);
        }

        config.clear_channels();
        CHECK_EQ(consts.recipes.count(config.recipe()), 1) << config.recipe();
        for (const auto& ch : consts.recipes.at(config.recipe())) {
          *config.add_channels() = ch;
        }
      } else {
        CHECK_THROW(!config.has_recipe()) << "You provided both recipe and channels.";
      }
    }

    // 4. light_bbo
    for (auto& [name, config] : *feed.mutable_exchanges()) {
      if (name.find("light_bbo") != std::string::npos ||
          config.recipe().find("light_bbo") != std::string::npos) {
        exchange::feed::RawParserConfig light_bbo_override;
        light_bbo_override.set_enable_diff_book(false);
        light_bbo_override.set_publish_bbo_book_only(true);
        light_bbo_override.set_book_level(5);
        config.mutable_raw_parser()->MergeFrom(light_bbo_override);
      } else if (
          name.find("move2bp") != std::string::npos ||
          config.recipe().find("move2bp") != std::string::npos) {
        exchange::feed::RawParserConfig move2bp_override;
        move2bp_override.set_sampling_frequency_bps(2);
        move2bp_override.set_book_level(5);
        if (name.find("trade_move2bp") != std::string::npos ||
            config.recipe().find("trade_move2bp") != std::string::npos) {
          move2bp_override.set_trade_sampling_frequency_bps(2);
        }
        config.mutable_raw_parser()->MergeFrom(move2bp_override);
      }
    }

    // 5. channels => websocket.channels, raw_parser.channels
    for (auto& [name, config] : *feed.mutable_exchanges()) {
      if (!config.channels().empty()) {
        if (config.websocket().channels().empty()) {
          config.mutable_websocket()->mutable_channels()->MergeFrom(config.channels());
        }
        if (config.raw_parser().channels().empty()) {
          config.mutable_raw_parser()->mutable_channels()->MergeFrom(config.channels());
        }
      }
    }
  }

  if (driver->has_strategy()) {
    auto& strategy = *driver->mutable_strategy();
    if (strategy.has_common()) {
      const auto& system_common = strategy.common();
      for (auto& [name, system_config] : *strategy.mutable_executors()) {
        if (!system_config.has_mea()) {
          system_config.set_mea(name);
        }
        system_config.MergeFrom(system_common);
      }
    }
    for (auto& [name, system_config] : *strategy.mutable_executors()) {
      // for this system_config,
      if (system_config.has_common()) {
        const auto& oe_common = system_config.common();
        for (auto& [group_name, group_config] : *system_config.mutable_products()) {
          group_config.MergeFrom(oe_common);
        }
      }
      for (auto& [group_name, group_config] : *system_config.mutable_products()) {
        if (!group_config.has_products()) {
          group_config.mutable_products()->add_norms(group_name);
        }
      }
    }
  }

  if (driver->has_order()) {
    auto& order = *driver->mutable_order();

    if (order.has_common()) {
      for (auto& [name, config] : *order.mutable_exchanges()) {
        auto c = order.common();
        c.MergeFrom(config);
        config = c;
      }
    }

    for (auto& [name, config] : *order.mutable_exchanges()) {
      if (!config.has_mea()) {
        config.set_mea(name);
      }
      CHECK_GT_THROW(name.size(), 0);
      CHECK_GT_THROW(config.mea().size(), 0);
    }

    for (auto& [name, config] : *order.mutable_exchanges()) {
      // propagate key_filepath
      if (config.has_key_filepath()) {
        if (!config.connection_config().has_key_filepath()) {
          config.mutable_connection_config()->set_key_filepath(config.key_filepath());
        }
      }
    }
  }

  driver->set_is_conslidated_do_not_set_this_field(true);
  return *driver;
}

}  // namespace coin2::app
